package com.chencong.online.function;

import com.chencong.online.bean.PageViewCountBean;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.shaded.guava18.com.google.common.collect.*;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Map;

/**
 * @program: user-behavior-analysis-online
 * @ClassName HotPageTopNProcessFunc
 * @description: 将窗口内聚合数据进行排序
 * @author: chencong
 * @create: 2021-12-23 13:59
 **/
public class HotPageTopNProcessFunc extends KeyedProcessFunction<Long, PageViewCountBean, String> {
    //定义Map状态
    MapState<String, Long> pageViewCountMapState;
    //定义构造器
    private Integer topSize;

    public HotPageTopNProcessFunc(Integer topSize) {
        this.topSize = topSize;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //初始化状态
        pageViewCountMapState = getRuntimeContext().getMapState(new MapStateDescriptor<String, Long>("page-view-count-map-state", String.class, Long.class));
    }

    @Override
    public void processElement(PageViewCountBean value, KeyedProcessFunction<Long, PageViewCountBean, String>.Context ctx, Collector<String> out) throws Exception {
        //添加状态
        pageViewCountMapState.put(value.getUrl(), value.getCount());
        //注册定时器，触发窗口计算
        ctx.timerService().registerEventTimeTimer(value.getWindowEnd() + 1);
        //注册一个1分钟后的定时器，清空状态，主要用来处理迟到数据
        ctx.timerService().registerEventTimeTimer(value.getWindowEnd() + 60 * 1000L);
    }

    @Override
    public void onTimer(long timestamp, KeyedProcessFunction<Long, PageViewCountBean, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
        //判断是否到了关闭窗口的时间，如果到了，直接清空状态返回
        if (timestamp == ctx.getCurrentKey() + 60 * 1000L) {
            pageViewCountMapState.clear();
        }
        //将数据放在list集合进行排序
        ArrayList<Map.Entry<String, Long>> pageViewCounts = Lists.newArrayList(pageViewCountMapState.entries());
        //将list集合进行排序
        pageViewCounts
                .sort(new Comparator<Map.Entry<String, Long>>() {
                    @Override
                    public int compare(Map.Entry<String, Long> o1, Map.Entry<String, Long> o2) {
                        if (o2.getValue() > o1.getValue()) {
                            return 1;
                        } else if (o2.getValue() < o1.getValue()) {
                            return -1;
                        } else {
                            return 0;
                        }
                    }
                });

        // 格式化成String输出
        StringBuilder resultBuilder = new StringBuilder();
        resultBuilder.append("===================================\n");
        resultBuilder.append("窗口结束时间：").append(new Timestamp(timestamp - 1)).append("\n");

        // 遍历列表，取top n输出
        for (int i = 0; i < Math.min(topSize, pageViewCounts.size()); i++) {
            Map.Entry<String, Long> currentItemViewCount = pageViewCounts.get(i);
            resultBuilder.append("NO ").append(i + 1).append(":")
                    .append(" 页面URL = ").append(currentItemViewCount.getKey())
                    .append(" 浏览量 = ").append(currentItemViewCount.getValue())
                    .append("\n");
        }
        resultBuilder.append("===============================\n\n");

        // 控制输出频率
        Thread.sleep(1000L);

        out.collect(resultBuilder.toString());

    }
}
